package com.we.mexico.flink.utils;

import org.apache.flink.api.java.utils.ParameterTool;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.io.InputStream;
import java.util.Set;

public class WeKafkaPropertyReader {
    private static ParameterTool param;
    public static Logger LOG = LoggerFactory.getLogger(WeKafkaPropertyReader.class);

    public static WeKafkaPropertyReader init(String propFilePath) {
        try {
            param = getReader(propFilePath);
        } catch (Exception e) {
            LOG.error("Exception: " + e.getMessage());
            e.printStackTrace();
        }
        return new WeKafkaPropertyReader();
    }

    public String getKfkBootStrapServer() {
        return param.get("kfkBootStrapServer");
    }

    public String getKfkTopic() {
        return param.get("kfkTopic");
    }

    public int getKfkPartitions() {
        return Integer.parseInt(param.get("kfkPartitions"));
    }

    public String getTmpKfkBootStrapServer() {
        return param.get("tmpkfkBootStrapServer");
    }

    public String getTmpKfkTopic() {
        return param.get("tmpkfkTopic");
    }

    public int getTmpKfkPartitions() {
        return Integer.parseInt(param.get("tmpkfkPartitions"));
    }

    public String getTmpKfkGroupId() {
        return param.get("tmpkfkGroupId");
    }

    public String getTmpKfkOffset() {
        return param.get("tmpkfkOffset");
    }

    public String getRocksDBBackendUrl() {
        return param.get("rocksDBBackendURL");
    }

    public String getHiveCatlogName() {
        return param.get("hiveCatlogName");
    }

    public String getHiveDBName() {
        return param.get("hiveDBName");
    }

    public String getHiveConfDir() {
        return param.get("hiveConfDir");
    }

    public static ParameterTool getReader(String filePath) throws IOException {
        InputStream resourceAsStream =
                WeKafkaPropertyReader.class.getClassLoader().getResourceAsStream(filePath);
        return ParameterTool.fromPropertiesFile(resourceAsStream);
    }

    public static void main(String[] args) {
        //        PropertyReader reader = PropertyReader.init("self.properties");
        //
        //        System.out.println(reader.getHiveCatlogName());
        //        System.out.println(reader.getHiveDBName());
        //        System.out.println(reader.getHiveConfDir());

        ParameterTool reader = null;
        try {
            reader = WeKafkaPropertyReader.getReader("risk/phonerecord.properties");
            System.out.println("size: " + reader.getProperties().size());
            Set<Object> keySet = reader.getProperties().keySet();

            System.out.println(
                    keySet.contains("fix_inbook_rejected_continous_uncall_3days_num_3m"));
            //            Iterator<Object> iterator = keySet.iterator();
            //            while(iterator.hasNext()){
            //                System.out.println(iterator.next());
            //            }

        } catch (IOException e) {
            e.printStackTrace();
        }

        //        ParameterTool parameterTool = ParameterTool.fromArgs(args);
        //        System.out.println(parameterTool.getNumberOfParameters());
        //        System.out.println("-hfqaudithomepagelog \n" +
        //                "-lendauditrecord \n" +
        //                "-offlinelabel \n" +
        //                "-risktag \n" +
        //                "-savemoneycardmember \n" +
        //                "-savemoncardmemspecial \n" +
        //                "-supermemrecord \n" +
        //                "-txxyorderrescallback \n" +
        //                "-useraccount \n" +
        //                "-useraudit \n" +
        //                "-usercertifycmpy \n" +
        //                "-usercertifyedu \n" +
        //                "-userentrepreneur \n" +
        //                "-userlevel \n" +
        //                "-userloan \n");
        //        System.out.println(parameterTool.has("testkey"));
        //        System.out.println(parameterTool.has("testv"));

    }
}
